package leo.peersim.paxos;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Vector;

import leo.peersim.common.LogicalClock;

import org.apache.log4j.Logger;

import peersim.core.CommonState;

/**
 * Learner agent
 * 
 * @author leo
 * @version June 19, 2009
 */
public class Learner extends PaxosAgent {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	private HashMap<LogicalClock, ReplicaCommand> _decidedBallot = new HashMap<LogicalClock, ReplicaCommand>();
	private Vector<ReplicaCommand> _holdBackQueue = new Vector<ReplicaCommand>();
	
	private long _nextExpectedSequenceNumber = 1;
	
	private String _name = "Learner";

	/**
	 * 
	 * @param container
	 */
	protected Learner(Paxos container) {
		super(container);
	}
	
	@Override
	protected String getName() {
		return this._name;
	}
	
	/**
	 * 
	 * @param d
	 */
	protected ReplicaCommand manage(Decide d) {
		
		if (!_decidedBallot.containsKey(d.getBallotId())) {
			
			ReplicaCommand rcom = new ReplicaCommand(d.getDecidedValue(), d.getOpName());
			
			this._decidedBallot.put(d.getBallotId(), rcom);
			this._holdBackQueue.add(rcom);
			
			logger.debug(pp() + "command _" + d.getOpName() + "_ with ballot " + d.getBallotId() + " has been decided with value _" + d.getDecidedValue() + "_");
			
			this.updateDeliveryQueue();
			logger.debug(pp() + "executed commands:" + container.getExecutedCommandHistory() + ", holdback: " + this._holdBackQueue);
				
		} else {
			// already decided, dropping message
		}
		
		return this._decidedBallot.get(d.getBallotId());
		
		
	}
	
	/**
	 * 
	 */
	private void updateDeliveryQueue() {
		
		Collections.sort(this._holdBackQueue, new ReplicaCommandComparator());
		//element at the head has the smaller sequence number 
		
		while(this._holdBackQueue.get(0).getSequenceNumber()==this._nextExpectedSequenceNumber) {
			
			ReplicaCommand rc = this._holdBackQueue.remove(0);
			this._nextExpectedSequenceNumber++;
			
			container.executeCommand(rc);
			
			if(this._holdBackQueue.isEmpty()) {
				break;
			}
		}
	}
}

class ReplicaCommand {
	
	private final long _decidedTime;
	private final String _commandName;
	private long _executionTime = -1;
	private final long sequenceNumber;

	public ReplicaCommand(long decidedValue, String opName) {
		this._commandName = opName;
		this._decidedTime = CommonState.getTime();
		this.sequenceNumber = decidedValue;
	}
	
	public void execute() {
		this._executionTime = CommonState.getTime();
	}

	public long getSequenceNumber() {
		return this.sequenceNumber;
	}
	
	public String toString() {
		return this._commandName + "@" + this.sequenceNumber;
	}

	public String getCommandName() {
		return this._commandName;
	}
	
	public long getExecutionTime() {
		return this._executionTime;
	}
	
	public long getDecidedTime() {
		return this._decidedTime;
	}
	
}

/**
 * <p>Defines a total order over {@link ReplicaCommand}.</p>
 * 
 * @author leo
 * @version June 20, 2009
 */
class ReplicaCommandComparator implements Comparator<ReplicaCommand> {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());

	/**
	 * 
	 * @return <tt>1</tt> if r1 has a sequence number > than r2, <tt>0</tt> if they
	 * have the same sequence number (note that if the invariance of the protocol is
	 * respected, it never happens), <tt>1</tt> otherwise.
	 */
	public int compare(ReplicaCommand r1, ReplicaCommand r2) {
		if (r1.getSequenceNumber()>r2.getSequenceNumber()) {
			return 1;
		} else if(r1.getSequenceNumber()<r2.getSequenceNumber()) {
			return -1;
		} else {
			logger.fatal("replica with same sequence number!");
			return 0;
		}
	}
	
}
